/**
 * Software License, Version 1.0
 *
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 *
 *
 *Redistribution and use in source and binary forms, with or without
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 *
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University.  For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering,
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 *
 *
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of
 * any other entity.  Indiana University disclaims any liability to any
 * recipient for claims brought by any other entity based on infringement of
 * intellectual property rights or otherwise.
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.reliable.impl;

import java.util.Arrays;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Properties;
import java.util.Vector;

import cgl.narada.event.EventHeaders;
import cgl.narada.event.EventID;
import cgl.narada.event.NBEvent;
import cgl.narada.event.NBEventTypes;
import cgl.narada.event.TemplateInfo;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.event.impl.NBEventGenerator;
import cgl.narada.matching.EntityProfileMatchingTrees;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.record.RecordService;
import cgl.narada.service.record.RecordServiceImpl;
import cgl.narada.service.reliable.CompanionDbInfo;
import cgl.narada.service.reliable.events.RdAckInvoiceEntityEvent;
import cgl.narada.service.reliable.events.RdAckResponseServiceInvoiceEvent;
import cgl.narada.service.reliable.events.RdArchivalServiceNotification;
import cgl.narada.service.reliable.events.RdCompanionEntityEvent;
import cgl.narada.service.reliable.events.RdNakInvoiceEntityEvent;
import cgl.narada.service.reliable.events.RdProfileUpdateRequest;
import cgl.narada.service.reliable.events.RdProfileUpdateResponse;
import cgl.narada.service.reliable.events.RdPublisherRecoveryInfo;
import cgl.narada.service.reliable.events.RdRecoveryEntityRequest;
import cgl.narada.service.reliable.events.RdRecoveryResponse;
import cgl.narada.service.reliable.events.RdRepublishedEntityEvent;
import cgl.narada.service.reliable.events.RdRetransmissionServiceEvent;
import cgl.narada.service.reliable.events.RdSubscriberRecoveryInfo;
import cgl.narada.service.replay.ReplayService;
import cgl.narada.service.storage.InventoryEvent;
import cgl.narada.service.storage.SequenceDestinations;
import cgl.narada.service.storage.StorageService;
import cgl.narada.service.storage.StorageServiceFactory;

/** This class provides a set of core functions that service implementations
 need to satisfy. All services within the system need to conform to this
 interface.

 This class pertains to the reliable delivery service being provided to
 entities within the system. The service is responsible for among other
 things
 (a) Maintain the list of registered entities
 (b) Load the appropriate matching engines, depending upon the type of
 templates being managed.
 (c) Store events received onto stable storage.
 (d) Manage invoice events and ensure efficient negotiations with
 publishing entities.

 @author Shrideep Pallickara and Hasan Bulut
 $Date$
 $Revision$
 */

public class ReliableDeliveryServiceImpl extends Thread implements
        NBEventListener {

    /** Maintains a list of all the registered entities */
    private Hashtable entities;

    /** Maintains a list of templates and the list of entities registered to this
     template. template-id's to a Hashtable of registered entities */
    private Hashtable templatesAndEntities;

    /** template-id's to templateInfo */
    private Hashtable templates;

    private String moduleName = "ReliableDeliveryServiceImpl: ";

    private Hashtable profiles;

    private Hashtable engines;

    private EntityProfileMatchingTrees entityMatchingTrees;

    private StorageService storageService = null;

    private int rdsEntityId;

    private RdCommunicationsMultiplexerImpl rdsMultiplexer;
    private RdsEventPercolator rdsPercolator;

    private ClientService clientService;
    private EventConsumer eventConsumer;
    private EventProducer eventProducer;

    private ReplayService replayService;

    private String rdsString = "ReliableDeliveryService/";

    private long sequenceNumber = 0;

    private Hashtable sequencesLastAssigned;

    private Object syncObject;
    private Object cmpObject;

    private boolean recovering = true;

    private boolean debug = true;

    private Hashtable junk;

    private RecordService recordService;

//    SessionRecorder sessionRecorder = SessionRecorder.getInstance();
//    SessionPlayer sessionPlayer = SessionPlayer.getInstance();
//    MetadataHandler metadataHandler = MetadataHandler.getInstance();

    Properties props;

    public ReliableDeliveryServiceImpl(int rdsEntityId,
                                       StorageService storageService,
                                       Properties props, String commType) throws
            ServiceException {
        this.props = props;
        this.rdsEntityId = rdsEntityId;
        this.storageService = storageService;
        initializeProducerConsumer(rdsEntityId, props, commType);
        setPriority(Thread.MAX_PRIORITY);
        rdsMultiplexer = new RdCommunicationsMultiplexerImpl();
        rdsMultiplexer.registerReliableDeliveryService(this);

        entities = new Hashtable();
        templates = new Hashtable();

        templatesAndEntities = new Hashtable();
        profiles = new Hashtable();
        sequencesLastAssigned = new Hashtable();

        entityMatchingTrees = new EntityProfileMatchingTrees();
        rdsPercolator = new RdsEventPercolator(this, storageService);

        syncObject = new Object();
        cmpObject = new Object();
        recoverFromFailure();

        replayService = ReplayService.getInstance();
        replayService.initialize(rdsEntityId, storageService,
                                 templatesAndEntities);

        recordService = RecordServiceImpl.getInstance();
        System.out.println("RecordServiceImpl: initializing");
        recordService.initialize(rdsEntityId, this, templatesAndEntities,
                                 entities, templates);
        System.out.println("RecordServiceImpl: initialized");

        junk = new Hashtable();
//        System.out.println("Starting SessionRecorder");
//        startSessionRecorder();
//        System.out.println("Starting SessionPlayer");
//        startSessionPlayer();
//        System.out.println("Starting MetadataHandler");
//        startMetadataHandler();
    }

// added
//    public void startSessionRecorder() {
//        props.put("transportType", "niotcp");
//        props.put("userName", "guest");
//        props.put("password", "password");
//
//        sessionRecorder.initialize(this, props);
//    }
//
//    public void startSessionPlayer() {
//        props.put("transportType", "niotcp");
//        props.put("userName", "guest");
//        props.put("password", "password");
//
//        sessionPlayer.initialize(this, props);
//    }
//
//    public void startMetadataHandler() {
//        props.put("transportType", "niotcp");
//        props.put("userName", "guest");
//        props.put("password", "password");
//
//        metadataHandler.initialize(props);
//
//    }

    public TemplateInfo createTemplateInfo(int templateId, int templateType,
                                           Object template) throws
            ServiceException {

        return clientService.createTemplateInfo(templateId, templateType,
                                                template);
    }

    public int getEntityId() {
        return rdsEntityId;
    }

    private void initializeProducerConsumer(int entityId, Properties props,
                                            String commType) throws
            ServiceException {
        clientService = SessionService.getClientService(entityId);
        clientService.initializeBrokerCommunications(props, commType);
        eventConsumer = clientService.createEventConsumer(this);
        eventProducer = clientService.createEventProducer();
    }

    public void registerEntity(int entityId) throws ServiceException {
        Object entityKey = new Integer(entityId);
        if (entities.containsKey(entityKey)) {
            throw new ServiceException(moduleName + "Entity [" + entityId
                                       + "] already registered");
        }
        entities.put(entityKey, entityKey);

        if (!recovering) {
            storageService.storeRegisteredEntity(entityId);
        }
    }

    /** Removes an entity from the list of registered entities. Also for every
     template id cycle through the entity list and remove the entity if
     present */
    public void deRegisterEntity(int entityId) throws ServiceException {
        Object entityKey = new Integer(entityId);

        if (!entities.containsKey(entityKey)) {
            throw new ServiceException(moduleName + "Entity [" + entityId
                                       + "] not registered with Service");
        }

        /** For every template id cycle through the entity list and remove
         the entity if present */
        Enumeration lists = templatesAndEntities.elements();

        while (lists.hasMoreElements()) {
            Hashtable entityList = (Hashtable) lists.nextElement();

            if (entityList.containsKey(entityKey)) {
                entityList.remove(entityKey);
            }

        }
        /* end while */

        storageService.removeRegisteredEntity(entityId);
    }

    /** Registers an entity with a managed template. */
    public void registerEntityForTemplate(int entityId, int templateId) throws
            ServiceException {
        Object entityKey = new Integer(entityId);
        Object templateKey = new Integer(templateId);

        if (!entities.containsKey(entityKey)) {
            throw new ServiceException(moduleName + "Entity [" + entityId
                                       + "] not registered with Service");
        }

        if (!templates.containsKey(templateKey)) {
            throw new ServiceException(moduleName + "Unknown template " +
                                       templateKey);
        }

        Hashtable entityList = (Hashtable) templatesAndEntities.get(templateKey);

        if (entityList.containsKey(entityKey)) {
            throw new ServiceException(moduleName + "Entity [" + entityId
                                       + "] already registered with template [" +
                                       templateId + "]");
        }

        entityList.put(entityKey, entityKey);
        System.out.println("\nAfter registering " + entityId + " with " +
                           templateId + "\nTemplatesAndEntities:\n" +
                           templatesAndEntities.toString());
        if (!recovering) {
            storageService.storeEntityForTemplate(entityId, templateId);
        }
    }

    /** Deregister's an entity from a managed template */
    public void deregisterEntityFromTemplate(int entityId, int templateId) throws
            ServiceException {
        System.out.println(moduleName + " Deregistering " + entityId + " from " +
                           templateId);
        Object entityKey = new Integer(entityId);
        Object templateKey = new Integer(templateId);

        if (!entities.containsKey(entityKey)) {
            throw new ServiceException(moduleName + "Entity [" + entityId
                                       + "] not registered with Service");
        }

        if (!templates.containsKey(templateKey)) {
            throw new ServiceException(moduleName + "Unknown template " +
                                       templateKey);
        }

        System.out.println("\nBefore de-registering " + entityId + " with " +
                           templateId + "\nTemplatesAndEntities:\n" +
                           templatesAndEntities.toString());

        Hashtable entityList = (Hashtable) templatesAndEntities.get(templateKey);
        if (!entityList.containsKey(entityKey)) {
            throw new ServiceException(moduleName + "Entity [" + entityId
                                       + "] not registered with template [" +
                                       templateId + "]");
        }

        entityList.remove(entityKey);
        System.out.println("\nAfter de-registering " + entityId + " with " +
                           templateId + "\nTemplatesAndEntities:\n" +
                           templatesAndEntities.toString());

        storageService.removeEntityFromTemplate(entityId, templateId);
    }

    public void addTemplateManagement(TemplateInfo templateInfo) throws
            ServiceException {
        int templateId = templateInfo.getTemplateId();
        Object templateKey = new Integer(templateId);

        if (templates.containsKey(templateKey)) {
            throw new ServiceException(moduleName + "Template [" + templateInfo
                                       + "] Previously registered!");
        }

        templates.put(templateKey, templateInfo);

        /** Initialize Hashtable for the management of registered entities */
        templatesAndEntities.put(templateKey, new Hashtable());

        if (!recovering) {
            storageService.storeTemplate(templateInfo);
        }

        sequencesLastAssigned.put(templateKey, new Long(0));

        registerSubscriptionsForTemplate(templateId);

        if (!recovering) {
            replayService.registerSubscriptionsForTemplate(templateId);
        }
    }

    public void removeTemplateManagement(int templateId) throws
            ServiceException {
        Object templateKey = new Integer(templateId);

        if (!templates.containsKey(templateKey)) {
            throw new ServiceException(moduleName + "Template [" + templateId
                                       + "] NOT Previously registered!");
        }

        templates.remove(templateKey);

        /** Remove entries pertaining to the management of registered entities */
        templatesAndEntities.remove(templateKey);

        storageService.removeTemplateManagement(templateId);

        replayService.deregisterSubscriptionsForTemplate(templateId);

    }

    /** Returns a array containing the list of managed templates */
    public TemplateInfo[] getListOfManagedTemplates() {
        return (TemplateInfo[]) templates.values().toArray(new TemplateInfo[0]);
    }

    /** Checks if an entity is registered to a certain template */
    private boolean isRegistered(int entityId, int templateId) throws
            ServiceException {
        Object entityKey = new Integer(entityId);
        Object templateKey = new Integer(templateId);

        if (!templates.containsKey(templateKey)) {
            throw new ServiceException(moduleName + "Template [" + templateId
                                       + "] NOT Previously registered!");
        }

        if (!entities.containsKey(entityKey)) {
            throw new ServiceException(moduleName + "Entity [" + entityId
                                       + "] not registered with Service");
        }

        Hashtable entityList = (Hashtable) templatesAndEntities.get(templateKey);
        if (entityList.containsKey(entityKey)) {
            return true;
        }

        return false;
    }

    public void processProfileUpdateRequest(
            RdProfileUpdateRequest profileUpdateRequest) throws
            ServiceException {
        long requestId = profileUpdateRequest.getRequestId();
        Profile profile = profileUpdateRequest.getProfile();
        boolean addProfile = profileUpdateRequest.isAddProfile();
        int templateId = profileUpdateRequest.getTemplateId();
        int entityId = profileUpdateRequest.getEntityId();

        RdProfileUpdateResponse profileUpdateResponse;
        String additionalInfo = "";
        boolean success = true;
        try {
            if (addProfile) {
                addProfile(templateId, profile);
                additionalInfo += "Addition of profile successful";
            } else {
                removeProfile(templateId, profile);
                additionalInfo += "Removal of profile successful";
            }
        } catch (ServiceException serEx) {
            success = false;
            additionalInfo += serEx.toString();
        }
        /*end try-catch */
        profileUpdateResponse = new RdProfileUpdateResponse(
                profileUpdateRequest,
                success, additionalInfo);

        String sendToTopic = entityId + "/ProfileManagement";
        NBEvent nbEvent = eventProducer.generateEvent(
                TemplateProfileAndSynopsisTypes.STRING, sendToTopic,
                profileUpdateResponse.getBytes());

        NBEventGenerator.setEventType(nbEvent,
                                      NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);

        eventProducer.publishEvent(nbEvent);
    }

    public synchronized void addProfile(int templateId, Profile profile) throws
            ServiceException {
        Object templateKey = new Integer(templateId);
        int entityId = profile.getDestination();

        /** Check to see if entity is authorized on template id */
        if (!isRegistered(entityId, templateId)) {
            throw new ServiceException(moduleName + "Entity [ " + entityId
                                       + "] not registered to template " +
                                       templateId);
        }

        TemplateInfo templateInfo = (TemplateInfo) templates.get(templateKey);

        if (templateInfo.getTemplateType() != profile.getProfileType()) {
            throw new ServiceException(moduleName + "Template Type["
                                       + templateInfo.getTemplateType() +
                                       " NOT " + "the same as"
                                       + profile.getProfileType());
        }

        Object profileKey = profile.getProfileId();

        if (profiles.containsKey(profileKey)) {
            throw new ServiceException(moduleName + "Profile [" + profile
                                       + "] was previously registered!");
        }

        manageProfile(profile, true);
        profiles.put(profileKey, profile);

        if (!recovering) {
            storageService.storeProfile(templateId, profile);
        }

    }

    /** Manage the addition or removal of a profile from the stored matching
     engines */
    private void manageProfile(Profile profile, boolean addProfile) throws
            ServiceException {
        entityMatchingTrees.manageSubscriptionProfile(profile, addProfile);
    }

    public void removeProfile(int templateId, Profile profile) throws
            ServiceException {
        Object templateKey = new Integer(templateId);
        int entityId = profile.getDestination();

        /** Check to see if entity is authorized on template id */
        if (!isRegistered(entityId, templateId)) {
            throw new ServiceException(moduleName + "Entity [ " + entityId
                                       + "] not registered to template " +
                                       templateId);
        }

        TemplateInfo templateInfo = (TemplateInfo) templates.get(templateKey);

        if (templateInfo.getTemplateType() != profile.getProfileType()) {
            throw new ServiceException(moduleName + "Template Type["
                                       + templateInfo.getTemplateType() +
                                       " NOT " + "the same as"
                                       + profile.getProfileType());
        }

        Object profileKey = profile.getProfileId();

        if (!profiles.containsKey(profileKey)) {
            throw new ServiceException(moduleName + "Profile [" + profile
                                       + "] was not previously registered!");
        }

        manageProfile(profile, false);
        profiles.remove(profileKey);

        storageService.removeProfile(templateId, profile);
    }

    /** Processes an ACK invoice event received from an entity */
    public void processAckInvoiceEvent(RdAckInvoiceEntityEvent ackInvoiceEvent) throws
            ServiceException {
        if (debug) {
            System.out.println(moduleName + "Received ACK Invoice event "
                               + ackInvoiceEvent);
        }

        boolean singleAck = ackInvoiceEvent.containsSingleAck();
        int entityId = ackInvoiceEvent.getEntityId();
        int templateId = ackInvoiceEvent.getTemplateId();
        long ackSeqNum = 0;
        long[] ackSeqNums = null;

        long entitySync = storageService.getSyncpoint(templateId, entityId);

        if (singleAck) {
            ackSeqNum = ackInvoiceEvent.getAckSequenceNumber();
            storageService.processAcknowledgement(ackSeqNum, entityId);
        } else {
            ackSeqNums = ackInvoiceEvent.getAckSequenceNumbers();
            for (int i = 0; i < ackSeqNums.length; i++) {
                if (ackSeqNum < ackSeqNums[i]) {
                    ackSeqNum = ackSeqNums[i];
                }
                /** This is to ensure that we do not expend IO operations on processing
                 acknowledgements that were previously issued */
                if (entitySync < ackSeqNums[i]) {
                    storageService.processAcknowledgement(ackSeqNums[i],
                            entityId);
                }
            }
        }
        /*end if(singleAck) */

        if (debug) {
            System.out.println(moduleName + "Retrieving missed sequences for "
                               + "template [" + templateId +
                               "] between sequence num [" + entitySync
                               + "] and [" + ackSeqNum + "]");
        }

        long[] missedSequences = storageService.retrieveMissedSequences(
                entitySync,
                ackSeqNum, templateId, entityId);
        long advanceSync = entitySync;

        if (missedSequences == null) {
            advanceSync = ackSeqNum;
        } else {
            /** If there are multiple acks, make sure that sync is advanced to the
             ackSeqNum that is smaller than the lowest missed sequence.
             If you ack 12, 19, 22 and missed 18, 21 --> sync is 12 */
            if (!singleAck) {
                Arrays.sort(missedSequences);
                Arrays.sort(ackSeqNums);
                long lowestMissedSequences = missedSequences[0];
                for (int i = 0; i < ackSeqNums.length; i++) {
                    if (ackSeqNums[i] < lowestMissedSequences) {
                        advanceSync = ackSeqNums[i];
                    }
                }
            }
        }
        /* if missedSequences == null */

        /** If the syncpoint associated with the entity has changed update it
         at the storage service */
        if (entitySync != advanceSync) {
            storageService.advanceSyncpoint(advanceSync, templateId, entityId);
        }
        RdAckResponseServiceInvoiceEvent ackResponse = new
                RdAckResponseServiceInvoiceEvent(
                        templateId, entityId, advanceSync);

        if (debug) {
            System.out.println(moduleName + "Advance SYNC for entityId=" +
                               entityId
                               + " to => " + advanceSync);
        }

        if (missedSequences != null) {
            if (missedSequences.length == 1) {
                ackResponse.setMissedSequence(missedSequences[0]);
            } else {
                ackResponse.setMissedSequences(missedSequences);
            }
            /* end missedSequence.length == 1*/
        }
        /* end (missedSequence != null) */

        /** Proceed to issue the ackResponse event to the entity id*/
        issueAckResponse(ackResponse);

    }

    /** Processes an nak invoice event received from an entity */
    public void processNakInvoiceEvent(RdNakInvoiceEntityEvent nakInvoiceEvent) throws
            ServiceException {
        boolean singleNak = nakInvoiceEvent.containsSingleNak();
        int entityId = nakInvoiceEvent.getEntityId();
        int templateId = nakInvoiceEvent.getTemplateId();
        long retransmissionId = nakInvoiceEvent.getRetransmissionId();
        long nakSeqNum = 0;
        long[] nakSeqNums = null;

        if (singleNak) {
            nakSeqNum = nakInvoiceEvent.getNakSequenceNumber();
            manageRetransmissionFor(entityId, nakSeqNum, retransmissionId);
        } else {
            nakSeqNums = nakInvoiceEvent.getNakSequenceNumbers();
            for (int i = 0; i < nakSeqNums.length; i++) {
                manageRetransmissionFor(entityId, nakSeqNums[i],
                                        retransmissionId);
            }
        }
        /* end (singleNak) */
    }

    /** Manage the retransmission requests for a given sequence number */
    private void manageRetransmissionFor(int entityId, long seqNum,
                                         long retransmissionId) throws
            ServiceException {
        controlMemoryUtilization();
        InventoryEvent inventoryEvent = storageService.getStoredEvent(seqNum);
        int templateId = inventoryEvent.getTemplateId();
        long seq = inventoryEvent.getSequenceNumber();
        long prevSeq = inventoryEvent.getPreviousSequenceNumber();
        EventID eventId = inventoryEvent.getEvent().getEventHeaders().
                          getEventId();

        RdArchivalServiceNotification archNotify = new
                RdArchivalServiceNotification(
                        templateId, eventId, seq, prevSeq);

        RdRetransmissionServiceEvent retransmit = new
                                                  RdRetransmissionServiceEvent(
                retransmissionId, inventoryEvent.getEvent(), archNotify);

        System.out.println(moduleName + "Issuing nakResponse to entity=" +
                           entityId
                           + ", for templateId=" + templateId +
                           ", and sequenceNumber = " + seq);
        //printMemoryUtilization();

        NBEvent nbEvent = eventProducer.generateEvent(
                TemplateProfileAndSynopsisTypes.STRING, entityId + "/Invoice",
                retransmit.getBytes());

        NBEventGenerator.setEventType(nbEvent,
                                      NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);
        eventProducer.publishEvent(nbEvent);

        /** To prevent memory utilization explosions */
        inventoryEvent = null;
        retransmit = null;
        nbEvent = null;
    }

    /** This method is used to control memory utlization. The method will not
     return until the memory utilization has actually gone down. */
    private void controlMemoryUtilization() {
        long totalMemory = 0;
        long UTIL_LIMIT = 50000000;

        totalMemory = Runtime.getRuntime().totalMemory();

        if (totalMemory > UTIL_LIMIT / 3) {
            System.gc();
        }

        while (totalMemory > UTIL_LIMIT) {
            try {
                Thread _currentThread = Thread.currentThread();
                System.out.println(moduleName + "Will yield()->" +
                                   _currentThread
                                   + " to control Memory Utilization of [" +
                                   totalMemory + "] bytes");
                Thread.yield();
                Thread.sleep(30);
                totalMemory = Runtime.getRuntime().totalMemory();
            } catch (InterruptedException ie) {
                System.out.println(moduleName +
                                   "Error while sleeping to control "
                                   + "memory utilization.");
            }
            /* end try-catch */

        }
        /* end while */
    }

    private void printMemoryUtilization() {
        long totalMemory = Runtime.getRuntime().totalMemory();
        double totalMB = totalMemory * 0.000001;
        System.out.println(moduleName + "Memory Utilization = " + totalMB +
                           " MB");
    }

    /** Processes a published event received from an entity */
    public void processPublishedEvent(NBEvent nbEvent) {
        if (debug) {
            junk.put(nbEvent.getEventHeaders().getEventId(), new Long(System
                    .currentTimeMillis()));
        }
        rdsPercolator.processPublishedEvent(nbEvent);
    }

    /** Processes a companion event received for a previously published event */
    public void processCompanionEvent(RdCompanionEntityEvent companionEvent) {
        if (debug) {
            Long junkDelay = (Long) junk.get(companionEvent.getEventId());
            if (junkDelay != null) {
                System.out.println(moduleName + "Companion arrived after = "
                                   +
                                   (System.currentTimeMillis() -
                                    junkDelay.longValue()));
            }
        }
        rdsPercolator.processCompanionEvent(companionEvent);
    }

    /* Process a republished event */
    public void processRepublishedEvent(RdRepublishedEntityEvent
                                        republishedEvent) {
        rdsPercolator.processRepublishedEvent(republishedEvent);
    }

    public void processEntityRecoveryRequest(
            RdRecoveryEntityRequest recoveryRequest) throws ServiceException {
        System.out.println(moduleName + "Received " + recoveryRequest);

        int templateId = recoveryRequest.getTemplateId();
        int entityId = recoveryRequest.getEntityId();
        long recoveryId = recoveryRequest.getRecoveryRequestId();

        RdRecoveryResponse recoveryResponse;
        if (recoveryRequest.isPublisherRecovery()) {
            int publisherCatenation = storageService.getCatenation(templateId,
                    entityId);
            RdPublisherRecoveryInfo pubRecoveryInfo = new
                    RdPublisherRecoveryInfo(
                            templateId, entityId, publisherCatenation);
            recoveryResponse = new RdRecoveryResponse(templateId, entityId,
                    recoveryId, pubRecoveryInfo);
        } else {
            RdSubscriberRecoveryInfo subRecoveryInfo =
                    constructSubscriberRecoveryInfo(
                            templateId, entityId);
            recoveryResponse = new RdRecoveryResponse(templateId, entityId,
                    recoveryId, subRecoveryInfo);
        }

        issueRecoveryResponse(recoveryResponse);
    }

    /** This method constructs a subscriber recovery info once a templateId
     and eventId have been specified */
    private RdSubscriberRecoveryInfo constructSubscriberRecoveryInfo(
            int templateId, int entityId) throws ServiceException {
        long startTime, elapsedTime;

        startTime = System.currentTimeMillis();
        long entitySync = storageService.getSyncpoint(templateId, entityId);
        elapsedTime = System.currentTimeMillis() - startTime;
        System.out.println(moduleName + "Retrieved entitySync point in ("
                           + elapsedTime + ") mSecs");

        long seqNum = sequenceNumber;

        startTime = System.currentTimeMillis();
        long[] missedSequences = storageService.retrieveMissedSequences(
                entitySync,
                seqNum, templateId, entityId);
        long advanceSync = entitySync;

        elapsedTime = System.currentTimeMillis() - startTime;
        System.out.println(moduleName + "Retrieved missed sequences in ("
                           + elapsedTime + ") mSecs");

        if (missedSequences == null) {
            advanceSync = seqNum;
        }
        /* if missedSequences == null */

        /** If the syncpoint associated with the entity has changed update it
         at the storage service */
        if (entitySync != advanceSync) {
            storageService.advanceSyncpoint(advanceSync, templateId, entityId);
        }

        RdAckResponseServiceInvoiceEvent ackResponse = new
                RdAckResponseServiceInvoiceEvent(
                        templateId, entityId, advanceSync);

        if (missedSequences != null) {
            if (missedSequences.length == 1) {
                ackResponse.setMissedSequence(missedSequences[0]);
            } else {
                ackResponse.setMissedSequences(missedSequences);
            }
            /* end missedSequence.length == 1*/
        }
        /* end (missedSequence != null) */

        /** Now that we have constructed the ackResponse object, we must also
         proceed to construct the list of registered profiles associated with
         an entity */
        Profile[] _entityProfileList = null;
        Vector _entityProfiles = new Vector();
        Enumeration e = profiles.elements();
        while (e.hasMoreElements()) {
            Profile _prof = (Profile) e.nextElement();
            if (_prof.getDestination() == entityId) {
                _entityProfiles.addElement(_prof);
            }
        }

        if (_entityProfiles.size() != 0) {
            _entityProfileList = (Profile[]) _entityProfiles.toArray(new
                    Profile[0]);
        }

        RdSubscriberRecoveryInfo subRecoveryInfo = new RdSubscriberRecoveryInfo(
                templateId, entityId, ackResponse, _entityProfileList);
        System.out.println(moduleName + "Created " + subRecoveryInfo);
        return subRecoveryInfo;
    }

    public void processDiscoveryRequest() {

    }

    public void onEvent(NBEvent nbEvent) {
        int eventType = nbEvent.getEventType();

        if (eventType == NBEventTypes.NORMAL_EVENT) {
            if (debug) {
                System.out.println(moduleName + "Received NORMAL event \n\n");
            }
            processPublishedEvent(nbEvent);
            return;
        }

        if (eventType == NBEventTypes.RELIABLE_DELIVERY_EXCHANGE) {
            rdsMultiplexer.manageReliableDeliveryExchange(nbEvent);
            return;
        }

        System.out.println(moduleName + "Unknown event type(" + eventType
                           + ") received.");
    }

    /** end onEvent() */

    public void registerSubscriptionsForTemplate(int templateId) throws
            ServiceException {
        System.out.println(moduleName +
                           "Registering subscriptions for template ["
                           + templateId + "]");
        Object templateKey = new Integer(templateId);
        TemplateInfo templateInfo = (TemplateInfo) templates.get(templateKey);
        if (templateInfo == null) {
            throw new ServiceException(moduleName + "Unknown templateId ["
                                       + templateKey + "]");
        }

        Object templateSynopsis = templateInfo.getTemplate();
        int templateType = templateInfo.getTemplateType();

        String subscriptionString = rdsString + templateId;
        Profile profile = clientService.createProfile(
                TemplateProfileAndSynopsisTypes.STRING, subscriptionString);
        eventConsumer.subscribeTo(profile);

        Profile synopsisProfile = null;
        if (templateType == TemplateProfileAndSynopsisTypes.STRING
            ||
            templateType == TemplateProfileAndSynopsisTypes.REGULAR_EXPRESSIONS
            || templateType == TemplateProfileAndSynopsisTypes.TAG_VALUE_PAIRS
            || templateType == TemplateProfileAndSynopsisTypes.XML) {
            synopsisProfile = clientService.createProfile(templateType,
                    (String) templateSynopsis);
        }

        if (templateType == TemplateProfileAndSynopsisTypes.INTEGER) {
            int intSynopsis = Integer.parseInt(((String) templateSynopsis));
            synopsisProfile = clientService.createProfile(templateType,
                    new Integer(intSynopsis));
        }

        if (synopsisProfile != null) {
            eventConsumer.subscribeTo(synopsisProfile);
        }

        Profile profile3 = clientService
                           .createProfile(TemplateProfileAndSynopsisTypes.
                                          STRING,
                                          subscriptionString + "/Invoice");

        Profile profile4 = clientService.createProfile(
                TemplateProfileAndSynopsisTypes.STRING, subscriptionString
                + "/Recovery");
        Profile profile5 = clientService.createProfile(
                TemplateProfileAndSynopsisTypes.STRING, subscriptionString
                + "/ProfileManagement");

        eventConsumer.subscribeTo(profile3);
        eventConsumer.subscribeTo(profile4);
        eventConsumer.subscribeTo(profile5);
    }

    public void deregisterSubscriptionsForTemplate(int templateId) throws
            ServiceException {

    }

    public void eventReadyForStorage() {
        synchronized (syncObject) {
            syncObject.notify();
        }
        ;
        /*end synchronized*/
    }

    public CompanionDbInfo getCompanionDbInfo(int templateId) {
        synchronized (cmpObject) {
            Object templateKey = new Integer(templateId);
            long lastSequenceOnThisTemplate = ((Long) sequencesLastAssigned
                                               .get(templateKey)).longValue();
            sequenceNumber++;
            System.out.println("TemplateId: " + templateId + " seqNum: " +
                               sequenceNumber);
            /** Update the last sequence number associated with a given template */
            sequencesLastAssigned.remove(templateKey);
            sequencesLastAssigned.put(templateKey, new Long(sequenceNumber));
            CompanionDbInfo cmpInfo = new CompanionDbInfo(sequenceNumber,
                    lastSequenceOnThisTemplate);
            return cmpInfo;
        }
    }

    public void run() {
        try {
            while (true) {
                if (rdsPercolator.hasEventsAvailableForStorage()) {
                    CompanionDbInfo cmpInfo = rdsPercolator.getReadyEventInfo();
                    System.out.println(cmpInfo.toString());
                    NBEvent nbEvent = null;
                    if(cmpInfo.getEventId() != null) {
                        nbEvent = rdsPercolator.getReadyNBEvent(cmpInfo.
                                getEventId());
                    }
                    storeEventToStorage(cmpInfo, nbEvent);
                } else {
                    synchronized (syncObject) {
                        if (debug) {
                            System.out.println(moduleName +
                                               "Waiting for events to be stored");
                        }
                        syncObject.wait();
                        if (debug) {
                            System.out.println(moduleName +
                                               "Waking up to store event(s)!!");
                        }
                    }
                    /*end synchronized*/
                }
                /*end if(availableToStore) */
            }
            /* end while(true) */
        } catch (ServiceException serEx) {
            System.out.println(moduleName + serEx);
        } catch (InterruptedException ie) {
            System.out.println(moduleName + ie);
        }
        /*end try-catch */
    }
    /*end run() */

    public void storeEventToStorage(CompanionDbInfo companionDbInfo,
                                    NBEvent nbEvent) throws ServiceException {
        int status = companionDbInfo.getStatus();
        if (status == 1) {
            Hashtable destinations = entityMatchingTrees
                                     .computeEventDestinations(nbEvent);
            EventHeaders eventHeaders = nbEvent.getEventHeaders();
            int templateId = eventHeaders.getTemplateId();
            EventID eventId = eventHeaders.getEventId();
            long lastSequenceOnThisTemplate = companionDbInfo.
                                              getPrevSequenceNumber();
            long sequenceNo = companionDbInfo.getSequenceNumber();

            SequenceDestinations sequenceDestinations;
            int[] dests = null;
            if (destinations != null) {
                dests = new int[destinations.size()];
                Enumeration e = destinations.keys();
                int index = 0;
                while (e.hasMoreElements()) {
                    dests[index] = ((Integer) e.nextElement()).intValue();
                }
                /*end while */
            }
            /*end if */
            if (dests == null) {
                dests = new int[1];
                dests[0] = 0;
            }
            sequenceDestinations = new SequenceDestinations(sequenceNo, dests);
            InventoryEvent inventoryEvent = new InventoryEvent(sequenceNo,
                    lastSequenceOnThisTemplate, templateId, eventId, nbEvent);
            storageService.store(companionDbInfo, inventoryEvent,
                                 sequenceDestinations);

            /** Make sure to yank event off the release buffers */
            rdsPercolator.yankEventOffReleaseBuffers(companionDbInfo);
            // send ack
            RdArchivalServiceNotification archivalNotification = new
                    RdArchivalServiceNotification(templateId, eventId,
                                                  sequenceNo,
                                                  lastSequenceOnThisTemplate);
            issueArchivalNotification(nbEvent, dests, archivalNotification);
            nbEvent = null;

            return;
        } else if (status == 0) {
            storageService.store(companionDbInfo, null, null);
            // do not send ack
        } else { // valid == 2 or something else which indicates event is
            // not recoverable
            storageService.store(companionDbInfo, null, null);
        }
    }

    private void issueArchivalNotification(NBEvent storedEvent,
                                           int[] destinations,
                                           RdArchivalServiceNotification
                                           archNotification) throws
            ServiceException {
        if (debug) {
            System.out.println(moduleName + "Issuing archival notification "
                               + "with sequence number = " +
                               archNotification.getSequenceNumber());
        }
        Long junkDelay = (Long) junk
                         .get(storedEvent.getEventHeaders().getEventId());

        if (junkDelay != null) {
            System.out.println(moduleName + "Total arch delay = "
                               +
                               (System.currentTimeMillis() -
                                junkDelay.longValue()));
        }

        eventProducer.setSuppressRedistributionToSource(true);
        NBEvent nbEvent = eventProducer.generateEvent(storedEvent
                .getContentSynopsisType(), storedEvent.getContentSynopsis(),
                archNotification.getBytes());

        NBEventGenerator.setEventType(nbEvent,
                                      NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);
        eventProducer.publishEvent(nbEvent);
    }

    private void issueAckResponse(RdAckResponseServiceInvoiceEvent ackResponse) throws
            ServiceException {
        int entityId = ackResponse.getEntityId();
        int templateId = ackResponse.getTemplateId();
        if (debug) {
            System.out.println(moduleName + "Issuing ->" + ackResponse);
        }

        NBEvent nbEvent = eventProducer.generateEvent(
                TemplateProfileAndSynopsisTypes.STRING, entityId + "/Invoice",
                ackResponse.getBytes());

        NBEventGenerator.setEventType(nbEvent,
                                      NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);
        eventProducer.publishEvent(nbEvent);

    }

    private void issueRecoveryResponse(RdRecoveryResponse recoveryResponse) throws
            ServiceException {
        int entityId = recoveryResponse.getEntityId();
        int templateId = recoveryResponse.getTemplateId();
        String sendToTopic;

        if (recoveryResponse.hasPublisherRecoveryInfo()) {
            sendToTopic = entityId + "/Recovery/Publisher";
        } else {
            sendToTopic = entityId + "/Recovery/Subscriber";
        }

        System.out.println(moduleName + "Issuing Recovery Response to entity="
                           + entityId + ", for templateId=" + templateId);
        NBEvent nbEvent = eventProducer.generateEvent(
                TemplateProfileAndSynopsisTypes.STRING, sendToTopic,
                recoveryResponse
                .getBytes());

        NBEventGenerator.setEventType(nbEvent,
                                      NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);

        eventProducer.publishEvent(nbEvent);

    }

    /** This method is invoked the first time a stable storage starts up,
     which might be after a failure or after a prolonged shutdown. This
     method enables the storage to recover from failures etc. */
    public void recoverFromFailure() throws ServiceException {
        int[] _entities = storageService.getListOfRegisteredEntities();

        if (_entities == null) {
            System.out.println(moduleName +
                               "No entities were previously registered"
                               +
                               " initialization using storage service complete ");
            recovering = false;
            return;
        }

        int _numOfEntities = _entities.length;
        for (int i = 0; i < _numOfEntities; i++) {
            registerEntity(_entities[i]);
        }
        /* end for */

        TemplateInfo[] _templates = storageService.getListOfManagedTemplates();
        if (_templates == null) {
            System.out.println(moduleName +
                               "No templates were previously registered"
                               +
                               " initialization using storage service complete ");
            recovering = false;
            return;
        }

        if (_templates.length == 0) {
            System.out.println(moduleName +
                               "No templates were previously registered"
                               +
                               " initialization using storage service complete ");
            recovering = false;
            return;
        }

        sequenceNumber = storageService.getSequenceNumberLastAssigned();
        System.out.println(moduleName + "SequenceNumberLastAssigned="
                           + sequenceNumber);

        int _numOfTemplates = _templates.length;
        for (int i = 0; i < _numOfTemplates; i++) {
            int _templateId = _templates[i].getTemplateId();
            System.out.println(moduleName + "Recovering for TemplateID ["
                               + _templateId + "]");
            Object _templateKey = new Integer(_templateId);
            addTemplateManagement((TemplateInfo) _templates[i]);

            /** Update the sequence number last assigned for a given templateId */
            long _syncForTemplate = storageService
                                    .getSequenceNumberLastAssigned(_templateId);
            sequencesLastAssigned.put(_templateKey, new Long(_syncForTemplate));

            processEntitiesRegisteredToTemplate(_templateId);
            processProfilesRegisteredToTemplate(_templateId);
        }

        System.out.println("\n\n\n" + moduleName + "Recovery from Failure "
                           + "(If, indeed, there was one) complete \n\n");
        /** We have now completed the recovery process completely */
        recovering = false;
    }

    private void processEntitiesRegisteredToTemplate(int templateId) throws
            ServiceException {
        int[] _entities = storageService.getListOfRegisteredEntities(templateId);
        if (_entities == null) {
            System.out.println(moduleName +
                               "No entities registered for template ["
                               + templateId + "]");
            return;
        }
        int _numOfEntities = _entities.length;

        for (int i = 0; i < _numOfEntities; i++) {
            registerEntityForTemplate(_entities[i], templateId);
        }
    }

    private void processProfilesRegisteredToTemplate(int templateId) throws
            ServiceException {
        Profile[] _profiles = storageService
                              .getListOfRegisteredProfiles(templateId);

        if (_profiles == null) {
            System.out.println(moduleName +
                               "No profiles registered for template ["
                               + templateId + "]");
            return;
        }

        int _numOfProfiles = _profiles.length;
        for (int i = 0; i < _numOfProfiles; i++) {
            addProfile(templateId, _profiles[i]);
        }

    }

    //added
    public boolean templateExists(int templateId) {
        return templates.containsKey("" + templateId);
    }

    //added
    public boolean entityExists(int entityId) {
        return entities.containsKey("" + entityId);
    }

    public static void main(String[] args) {
        String module = "ReliableDeliveryServiceImpl.main() ->";
        int entityId = 123678;
        Properties props = new Properties();
        /** These properties pertain to setting up a TCP link */
        props.put("hostname", args[0]);
        props.put("portnum", args[1]);

        try {
            StorageService storageService = StorageServiceFactory.
                                            getStorageService(
                    "db", new Properties());
//    	StorageService storageService = new StorageServiceDbImpl();
            ReliableDeliveryServiceImpl rds = new ReliableDeliveryServiceImpl(
                    entityId, storageService, props, "niotcp");
            rds.start();

//      System.out.println(module + "REGISTERING Subscription ");
//      TemplateInfo templateInfo = rds.createTemplateInfo(12345,
//          TemplateProfileAndSynopsisTypes.STRING, "Movie/Casablanca");
//
//      if (args.length == 3) {
//        rds.addTemplateManagement(templateInfo);
//        rds.registerEntity(7777);
//        rds.registerEntity(7007);
//        rds.registerEntityForTemplate(7777, 12345);
//        rds.registerEntityForTemplate(7007, 12345);
//        System.out.println(module + "Registering " + templateInfo + "\n "
//            + "Entities [7777] and [7007]");
//
//        cgl.narada.matching.string.StringProfile profile = new cgl.narada.matching.string.StringProfile(
//            "Movie/Casablanca", 7777, "11235Casab");
//      }

            //rds.addProfile(12345, profile);
            // } catch (IOException ioe) {
            //System.out.println(ioe);
        } catch (ServiceException serEx) {
            System.out.println(serEx);
        }
        /*end try-catch */
    }

}
